package drds.global_transaction.transaction_coordinator.lock_manager;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import drds.global_transaction.api.exception.GlobalTransactionException;
import drds.global_transaction.transaction_coordinator.transaction_session.BranchTransactionSession;
import io.netty.util.internal.ConcurrentSet;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class LockManagerImpl implements LockManager
{

	private static final int bucket_num_per_table = 128;
	private static final ConcurrentHashMap<String, // resourceId
			ConcurrentHashMap<String, // tableName
					ConcurrentHashMap<Integer, // bucketId
							// lockRecordId->globalTransactionId
							Map<String, Long>>>> lockMap = new ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>>>();

	@Override
	public boolean acquireLock(BranchTransactionSession branchTransactionSession) throws GlobalTransactionException
	{
		String resourceId = branchTransactionSession.getResourceId();
		long globalTransactionId = branchTransactionSession.getGlobalTransactionId();
		//
		ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>> resourceLockMap = lockMap.get(resourceId);
		if (resourceLockMap == null)
		{
			lockMap.putIfAbsent(resourceId, new ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>>());
			resourceLockMap = lockMap.get(resourceId);
		}
		//

		String[] tableNameAndLockRecordIdsStrings = branchTransactionSession.getTableNameAndLockRecordIdsStringsString().split(";");
		for (String tableNameAndLockRecordIdsString : tableNameAndLockRecordIdsStrings)
		{
			int index = tableNameAndLockRecordIdsString.indexOf(":");
			String tableName = tableNameAndLockRecordIdsString.substring(0, index);
			String lockRecordIdsString = tableNameAndLockRecordIdsString.substring(index + 1);
			//
			ConcurrentHashMap<Integer, Map<String, Long>> tableLockMap = resourceLockMap.get(tableName);
			if (tableLockMap == null)
			{
				resourceLockMap.putIfAbsent(tableName, new ConcurrentHashMap<Integer, Map<String, Long>>());
				tableLockMap = resourceLockMap.get(tableName);
			}
			String[] lockRecordIds = lockRecordIdsString.split(",");
			for (String lockRecordId : lockRecordIds)
			{
				//
				int bucketId = lockRecordId.hashCode() % bucket_num_per_table;
				Map<String, Long> bucketLockMap = tableLockMap.get(bucketId);
				if (bucketLockMap == null)
				{
					tableLockMap.putIfAbsent(bucketId, new HashMap<String, Long>());
					bucketLockMap = tableLockMap.get(bucketId);
				}
				synchronized (bucketLockMap)
				{
					Long lockedGlobalTransactionId = bucketLockMap.get(lockRecordId);
					if (lockedGlobalTransactionId == null)
					{
						bucketLockMap.put(lockRecordId, globalTransactionId);
						//
						// 下面是建立双向关联
						ConcurrentHashMap<Map<String, Long>, Set<String>> bucketLockMapToLockRecordIdSetMap = branchTransactionSession.getBucketLockMapToLockRecordIdSetMap();
						Set<String> lockRecordIdSet = bucketLockMapToLockRecordIdSetMap.get(bucketLockMap);
						if (lockRecordIdSet == null)
						{
							bucketLockMapToLockRecordIdSetMap.putIfAbsent(bucketLockMap, new ConcurrentSet<String>());
							lockRecordIdSet = bucketLockMapToLockRecordIdSetMap.get(bucketLockMap);//
						}
						lockRecordIdSet.add(lockRecordId);

					} else if (lockedGlobalTransactionId.longValue() == globalTransactionId)
					{
						// locked by self
						continue;
					} else
					{
						// not equals
						branchTransactionSession.unlock(); // Release all acquired locks.
						return false;
					}
				}
			}
		}
		return true;
	}

	@Override
	public boolean isLockable(String resourceId, long globalTransactionId, String tableNameAndLockRecordIdsStringsString) throws GlobalTransactionException
	{
		ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>> resourceLockMap = lockMap.get(resourceId);
		if (resourceLockMap == null)
		{
			return true;
		}
		String[] tableNameAndLockRecordIdsStrings = tableNameAndLockRecordIdsStringsString.split(";");
		for (String tableNameAndLockRecordIdsString : tableNameAndLockRecordIdsStrings)
		{
			int index = tableNameAndLockRecordIdsString.indexOf(":");
			String tableName = tableNameAndLockRecordIdsString.substring(0, index);
			String lockRecordIdsString = tableNameAndLockRecordIdsString.substring(index + 1);
			ConcurrentHashMap<Integer, Map<String, Long>> tableLockMap = resourceLockMap.get(tableName);
			if (tableLockMap == null)
			{
				continue;
			}
			String[] lockRecordIds = lockRecordIdsString.split(",");
			for (String lockRecordId : lockRecordIds)
			{
				int bucketId = lockRecordId.hashCode() % bucket_num_per_table;
				Map<String, Long> bucketLockMap = tableLockMap.get(bucketId);
				if (bucketLockMap == null)
				{
					continue;
				}
				Long lockingTransactionId = bucketLockMap.get(lockRecordId);
				if (lockingTransactionId == null || lockingTransactionId.longValue() == globalTransactionId)
				{
					// locked by self
					continue;
				} else
				{
					return false;
				}
			}
		}
		return true;
	}
}
